package cn.dmp.test

import cn.dmp.util.{AppParams, Jpools, RedisHander}
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}
import redis.clients.jedis.Jedis

import scala.collection.mutable.ArrayBuffer

/**
  * 数据标签化
  */
object KpiImageMainServiceRedis {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
    conf.setAppName("数据标签化")
    conf.setMaster("local[*]")
    //序列化方式
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    val sc: SparkContext = new SparkContext(conf)

    val stopWords: RDD[String] = sc.textFile("H:\\bigdata\\大数据课程\\杨家伟视频\\项目二\\资料PDF\\stopwords.txt")
    val sotopWordBT: Broadcast[Array[String]] = sc.broadcast(stopWords.collect())

    val sQLContext = new SQLContext(sc)
    /**
      * adspacetype: Int, 广告位类型（1：banner 2：插屏 3：全屏）
      * adspacetypename: String, 广告位类型名称（banner、插屏、全屏
      * appid: String, 应用 id
      * appname: String, 应用名称
      * adplatformproviderid: Int, 	广告平台商id  (>= 100000: rtb)
      *
      * client: Int, 操作系统 （1：android 2：ios 3：wp）
      * networkmannername:String  联网方式名称
      * ispname: String, 运营商名称
      * keywords: String, 	关键字
      * provincename: String, 	设备所在省份名称
      * cityname: String, 	设备所在城市名称
      *
      * userId
      *
      * imei: String, 	imei
      * mac: String, 	mac
      * idfa: String, 	idfa
      * openudid: String, 	openudid
      * androidid: String, 	androidid
      * imeimd5: String, 	imei_md5
      * macmd5: String, 	mac_md5
      * idfamd5: String, 	idfa_md5
      * openudidmd5: String, 	openudid_md5
      * androididmd5: String,	androidid_md5
      * imeisha1: String, 	imei_sha1
      * macsha1: String, 	mac_sha1
      * idfasha1: String, 	idfa_sha1
      * openudidsha1: String, 	openudid_sha1
      * androididsha1: String,	androidid_sha1
      *
      */
    //联网方式
    //var clientMap = Map[Int, String](1 -> "D00010001", 2 -> "D00010002", 3 -> "D00010003")
    var clientMap =AppParams.clientMap

    //设备联网方式
    //val networkMap: Map[String, String] = Map[String, String]("WIFI" -> "D00020001", "4G" -> "D00020002", "3G" -> "D00020003", "2G" -> "D00020004")
    var networkMap =AppParams.networkMap

    //设备运营商方式
    //val ispnameMap: Map[String, String] = Map[String, String]("移动" -> "D00030001", "联通" -> "D00030002", "电信" -> "D00030003")
    var ispnameMap =AppParams.ispnameMap

    val frame: DataFrame = sQLContext.read.parquet("logFile")

    val logMap: RDD[(String, String, String, String, String, String, List[String], String, String)] = frame.map(row => {

      val adspacetype: Int = row.getAs[Int]("adspacetype")
      //1) 广告位类型
      var adspacetypeImage = "-1"
      if (!adspacetype.equals("0")) {
        adspacetypeImage = if (adspacetype < 10) "LC0" + adspacetype else "LC" + adspacetype
      }

      val adspacetypename: String = row.getAs[String]("adspacetypename")
      //1) 广告位类型 名称
      var adspacetypenameImage = "LN" + adspacetypename

      val appid: String = row.getAs[String]("appid")
      val appname: String = row.getAs[String]("appname")
      //redis获取appname
      //2) App 名称
      var appnameImage: String = RedisHander.getAppName(appname, appid)

      if (appnameImage.equals("0")) appnameImage = "-1" else appnameImage = "APP" + appnameImage


      val adplatformproviderid: Int = row.getAs[Int]("adplatformproviderid")
      //3) 渠道
      val adplatformprovideridImage = if (adplatformproviderid != 0) "CN" + adplatformproviderid else "-1"

      val client: Int = row.getAs[Int]("client")
      //4) 设备 设备操作系统

      val clientImage: String = clientMap.getOrElse(client.toString, "D00010004").toString


      val networkmannername: String = row.getAs[String]("networkmannername")
      //4) 设备 设备操作系统
      val networkImage: String = networkMap.getOrElse(networkmannername, "D00020005").toString

      val ispname: String = row.getAs[String]("ispname")
      //4) 设备 设备运营商方式
      val ispnameImage: String = ispnameMap.getOrElse("ispname", "D00030004").toString

      val keywords: String = row.getAs[String]("keywords")
      val buffer: ArrayBuffer[String] = new ArrayBuffer[String]()

      //5) 关键字
      val tbValue: Array[String] = sotopWordBT.value
      if (!keywords.equals("0") && keywords.length >= 3 && keywords.length <= 8) {
        if (keywords.indexOf("|") > 0) {
          val words: Array[String] = keywords.split("\\|")
          words.foreach(word => {
            if (!tbValue.contains(keywords) && keywords.length >= 3 && keywords.length <= 8) {
              buffer += "K" + word
            }
          })
        } else if(!tbValue.contains(keywords)){
          buffer += "K" + keywords
        }
      }
      //val keywordImage: String = if (buffer.size == 0) "-1" else buffer.mkString(":")


      val provincename: String = row.getAs[String]("provincename")

      //6) 地域标签（省标签格式
      val provincenameImage: String = if (!provincename.equals("0")) "ZP" + provincename else "-1"

      val cityname: String = row.getAs[String]("cityname")
      //6) 地域标签（地市签格式
      val citynameImage: String = if (!cityname.equals("0")) "ZC" + cityname else "-1"


      //需要得到 userId
      val imei: String = row.getAs[String]("imei")
      val mac: String = row.getAs[String]("mac")
      val idfa: String = row.getAs[String]("idfa")
      val openudid: String = row.getAs[String]("openudid")
      val androidid: String = row.getAs[String]("androidid")
      val imeimd5: String = row.getAs[String]("imeimd5")
      val macmd5: String = row.getAs[String]("macmd5")
      val idfamd5: String = row.getAs[String]("idfamd5")
      val openudidmd5: String = row.getAs[String]("openudidmd5")
      val androididmd5: String = row.getAs[String]("androididmd5")
      val imeisha1: String = row.getAs[String]("imeisha1")
      val macsha1: String = row.getAs[String]("macsha1")
      val idfasha1: String = row.getAs[String]("idfasha1")
      val openudidsha1: String = row.getAs[String]("openudidsha1")
      val androididsha1: String = row.getAs[String]("androididsha1")

      val userId: String = getUserId(imei, mac, idfa, openudid, androidid, imeimd5, macmd5, idfamd5, openudidmd5, androididmd5, imeisha1, macsha1, idfasha1, openudidsha1, androididsha1)

      (userId, adspacetypeImage, adspacetypenameImage, appnameImage, adplatformprovideridImage, ispnameImage, buffer.toList, provincenameImage, citynameImage)
    })


    logMap.foreachPartition(itr=>{
      val jedis: Jedis = Jpools.getJedis()

      itr.foreach(tp=>{
       if(!tp._1.equals("-1")){
         if(!tp._2.equals("-1")) jedis.hincrBy(tp._1,tp._2,1)
         if(!tp._3.equals("-1")) jedis.hincrBy(tp._1,tp._3,1)
         if(!tp._4.equals("-1")) jedis.hincrBy(tp._1,tp._4,1)
         if(!tp._5.equals("-1")) jedis.hincrBy(tp._1,tp._5,1)
         if(!tp._6.equals("-1")) jedis.hincrBy(tp._1,tp._6,1)
         if(!tp._8.equals("-1")) jedis.hincrBy(tp._1,tp._8,1)
         if(!tp._9.equals("-1")) jedis.hincrBy(tp._1,tp._9,1)
         tp._7.foreach(t=>{
           if(!t.equals("-1")) jedis.hincrBy(tp._1,t,1)
         })
       }
      })
      jedis.close()

    })

    sc.stop()
  }

  // 获取一个识别id
  def getUserId(imei: String,
                mac: String,
                idfa: String,
                openudid: String,
                androidid: String,
                imeimd5: String,
                macmd5: String,
                idfamd5: String,
                openudidmd5: String,
                androididmd5: String,
                imeisha1: String,
                macsha1: String,
                idfasha1: String,
                openudidsha1: String,
                androididsha1: String): String = {
    if (!imei.equals("0")) {
      "D:" + imei.toUpperCase
    } else if (!mac.equals("0")) {
      "D:" + mac.toUpperCase
    } else if (!idfa.equals("0")) {
      "D:" + idfa.toUpperCase
    } else if (!openudid.equals("0")) {
      "D:" + openudid.toUpperCase
    } else if (!androidid.equals("0")) {
      "D:" + androidid.toUpperCase
    } else if (!imeimd5.equals("0")) {
      "D:" + imeimd5.toUpperCase
    } else if (!macmd5.equals("0")) {
      "D:" + macmd5.toUpperCase
    } else if (!idfamd5.equals("0")) {
      "D:" + idfamd5.toUpperCase
    } else if (!openudidmd5.equals("0")) {
      "D:" + openudidmd5.toUpperCase
    } else if (!androididmd5.equals("0")) {
      "D:" + androididmd5.toUpperCase
    } else if (!imeisha1.equals("0")) {
      "D:" + imeisha1.toUpperCase
    } else if (!macsha1.equals("0")) {
      "D:" + macsha1.toUpperCase
    } else if (!idfasha1.equals("0")) {
      "D:" + idfasha1.toUpperCase
    } else if (!openudidsha1.equals("0")) {
      "D:" + openudidsha1.toUpperCase
    } else if (!androididsha1.equals("0")) {
      "D:" + androididsha1.toUpperCase
    } else {
      "-1"
    }
  }
}
